//package com.example.service;
//
//import com.example.Utils.Env;
//import com.example.entity.StockTrade;
//import jakarta.annotation.PostConstruct;
//import jakarta.annotation.Resource;
//import lombok.extern.slf4j.Slf4j;
//import org.apache.flink.api.common.functions.AggregateFunction;
//import org.apache.flink.api.common.functions.MapFunction;
//import org.apache.flink.api.java.tuple.Tuple2;
//import org.apache.flink.configuration.Configuration;
//import org.apache.flink.streaming.api.datastream.DataStream;
//import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
//import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
//import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
//import org.apache.flink.streaming.api.windowing.time.Time;
//import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
//import org.springframework.stereotype.Component;
//import redis.clients.jedis.Jedis;
//
//@Component
//@Slf4j
//public class TopTenStock {
//    @Resource
//    private Env env_;
//
//    @PostConstruct
//    public void init() {
//        new Thread(() -> {
//            try {
//                startFlinkJob();
//            } catch (Exception e) {
//                log.error("Error starting Flink job", e);
//            }
//        }).start();
//    }
//
//    public  void startFlinkJob() throws Exception {
//        // 创建 Kafka source
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//
//        DataStream<StockTrade> tradeStream = env_.getEnv(env);
//
//        log.info("数据源搞好了");
//
//        // 1 小时窗口
//        tradeStream
//                .map(new MapFunction<StockTrade, Tuple2<String, Double>>() {
//                    @Override
//                    public Tuple2<String, Double> map(StockTrade value) throws Exception {
//                        return new Tuple2<>(value.getStockCode(), value.getTradeVolume() * value.getPrice());
//                    }
//                })
//                .keyBy(value -> value.f0)
//                .window(TumblingEventTimeWindows.of(Time.hours(1)))
//                .aggregate(new TradeAggregator())
//                .addSink(new AccumulatingRedisSink(redisConfig));
//
//        System.out.println("Top 10 stock trade volume analysis started");
//        env.execute("Stock Trade Analysis");
//    }
//
//    public FlinkJedisPoolConfig redisConfig = new FlinkJedisPoolConfig.Builder()
//            .setHost("localhost")
//            .setPort(6379)
//            .build();
//
//    // 自定义 Sink 函数
//// 自定义 Sink 函数
//public static class AccumulatingRedisSink extends RichSinkFunction<Tuple2<String, Double>> {
//    private transient Jedis jedis;
//    private final FlinkJedisPoolConfig config;
//
//    public AccumulatingRedisSink(FlinkJedisPoolConfig config) {
//        this.config = config;
//    }
//
//    @Override
//    public void open(Configuration parameters) throws Exception {
//        super.open(parameters);
//        jedis = new Jedis(config.getHost(), config.getPort());
//    }
//
//    @Override
//    public void close() throws Exception {
//        if (jedis != null) {
//            jedis.close();
//        }
//        super.close();
//    }
//
//@Override
//public void invoke(Tuple2<String, Double> value, Context context) throws Exception {
//    final long delayInterval = 100; // 每次写入之间的延迟时间（毫秒）
//
//    String hashKey = "stock_trade_volumes"; // 存储股票交易量的键
//    String field = value.f0; // 使用股票代码作为字段
//
//    // 检查键的类型
//    String type = jedis.type(hashKey);
//    if (!"zset".equals(type)) {
//        // 如果键不存在或类型不正确，删除旧的键
//        jedis.del(hashKey);
//        // 创建新的有序集合
//        jedis.zadd(hashKey, 0.0, field); // 初始化一个空的有序集合
//    }
//
//    // 从 Redis 中获取现有值
//    Double existingVolume = jedis.zscore(hashKey, field);
//    if (existingVolume == null) {
//        existingVolume = 0.0;
//    }
//
//    // 更新交易量
//    double updatedVolume = existingVolume + value.f1;
//
//    // 将更新后的值写回 Redis
//    jedis.zadd(hashKey, updatedVolume, field);
//    Thread.sleep(delayInterval);
//}
//
//}
//
//
//    // 定义聚合函数
//    public static class TradeAggregator implements AggregateFunction<Tuple2<String, Double>, Tuple2<String, Double>, Tuple2<String, Double>> {
//
//
//        @Override
//        public Tuple2<String, Double> createAccumulator() {
//            return new Tuple2<>("", 0.0);
//        }
//
//        @Override
//        public Tuple2<String, Double> add(Tuple2<String, Double> stringDoubleTuple2, Tuple2<String, Double> stringDoubleTuple22) {
//            String stockCode = stringDoubleTuple2.f0;
//            double totalAmount = stringDoubleTuple2.f1 + stringDoubleTuple22.f1;
//            return new Tuple2<>(stockCode, totalAmount);
//        }
//
//
//
//
//        @Override
//        public Tuple2<String, Double> getResult(Tuple2<String, Double> accumulator) {
//            return accumulator;
//        }
//
//        @Override
//        public Tuple2<String, Double> merge(Tuple2<String, Double> a, Tuple2<String, Double> b) {
//            String stockCode = a.f0; // 假设两个 accumulator 的股票代码相同
//            double totalAmount = a.f1 + b.f1;
//            return new Tuple2<>(stockCode, totalAmount);
//        }
//
//
//    }
//}
